/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal/allocator.h"
#include "dal/client.h"
#include "dal_fastdds/utils/log.hpp"
#include "dal_fastdds/utils/EntityGetter.hpp"
#include "dal_fastdds/utils/ClientResponse.hpp"
#include "dal_fastdds/converter/QosConverter.hpp"
#include "dal_fastdds/entity_impl/DalClientImpl.hpp"
#include "dal_fastdds/config_manager/ConfigManager.hpp"
#include "dal_fastdds/listener_impl/ClientRespReaderListenerImpl.hpp"
#include "dal_fastdds/listener_impl/ClientReqWriterListenerImpl.hpp"
#include "dal_utils/MsgLibLoader.hpp"

using namespace eprosima::fastdds::dds;
using namespace eprosima::fastrtps::rtps;
using namespace dal_fastdds;

static dal_client_t* CreateDalClientByQos(dal_domain_participant_t* participant,
                                          const char* service_name,
                                          const char* request_writer_type_name,
                                          const char* response_reader_type_name,
                                          DataWriterQos& f_wqos,
                                          DataReaderQos& f_rqos)
{
    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return nullptr;
    }

    const auto& dal_part_impl = reinterpret_cast<DalDomainParticipantImpl*>(participant->handle_);
    if (!dal_part_impl->CreateCSPubSub()) {
        LOG_ERROR_WITH_LOCATION() << "CreateCSPubSub failed." << LOG_ENDL;
        return nullptr;
    }

    auto req_writer_ts = static_cast<const message_type_support_callbacks_t*>(
        MsgLibLoader::typeSupportMap_[request_writer_type_name]);
    auto resp_reader_ts = static_cast<const message_type_support_callbacks_t*>(
        MsgLibLoader::typeSupportMap_[response_reader_type_name]);
    if (!req_writer_ts || !resp_reader_ts) {
        LOG_ERROR_WITH_LOCATION() << "req writer or resp reader type support not found."
                                  << LOG_ENDL;
        return nullptr;
    }

    // find or create topics
    DalClientImpl* client_impl = new DalClientImpl(service_name, req_writer_ts, resp_reader_ts);
    client_impl->SetParticipant(f_participant);
    client_impl->SetCSPublisher(dal_part_impl->CSPublisher());
    client_impl->SetCSSubscriber(dal_part_impl->CSSubscriber());

    TypeSupport req_type;
    TopicDescription* req_topic_desc = nullptr;
    if (!client_impl->GetFastddsReqType(req_topic_desc, req_type)) {
        LOG_ERROR_WITH_LOCATION() << "get request type failed." << LOG_ENDL;
        return nullptr;
    }

    TypeSupport resp_type;
    TopicDescription* resp_topic_desc = nullptr;
    if (!client_impl->GetFastddsRespType(resp_topic_desc, resp_type)) {
        LOG_ERROR_WITH_LOCATION() << "get response type failed." << LOG_ENDL;
        return nullptr;
    }

    if (ReturnCode_t::RETCODE_OK != req_type.register_type(f_participant)) {
        LOG_ERROR_WITH_LOCATION() << "register req type failed." << LOG_ENDL;
        return nullptr;
    }
    if (ReturnCode_t::RETCODE_OK != resp_type.register_type(f_participant)) {
        LOG_ERROR_WITH_LOCATION() << "register resp type failed." << LOG_ENDL;
        return nullptr;
    }
    client_impl->req_fastdds_type_ = req_type;
    client_impl->resp_fastdds_type_ = resp_type;

    if (!client_impl->GetFastddsReqTopic(req_topic_desc) ||
        !client_impl->GetFastddsRespTopic(resp_topic_desc)) {
        LOG_ERROR_WITH_LOCATION() << "get req or resp type failed." << LOG_ENDL;
        return nullptr;
    }

    dal_part_impl->AddCSTopic(dynamic_cast<Topic*>(req_topic_desc));
    dal_part_impl->AddCSTopic(dynamic_cast<Topic*>(resp_topic_desc));

    // create listener
    client_impl->resp_reader_listener_ = new ClientRespReaderListenerImpl(client_impl);
    if (!client_impl->resp_reader_listener_) {
        LOG_ERROR_WITH_LOCATION() << "create resp reader listener failed." << LOG_ENDL;
        return nullptr;
    }

    client_impl->req_writer_listener_ = new ClientReqWriterListenerImpl(client_impl);
    if (!client_impl->req_writer_listener_) {
        LOG_ERROR_WITH_LOCATION() << "create req writer listener failed." << LOG_ENDL;
        return nullptr;
    }

    // create req_writer
    DataWriter* req_writer =
        dal_part_impl->CSPublisher()->create_datawriter(dynamic_cast<Topic*>(req_topic_desc),
                                                        f_wqos,
                                                        client_impl->req_writer_listener_,
                                                        StatusMask::publication_matched());
    if (!req_writer) {
        LOG_ERROR_WITH_LOCATION() << "create request writer failed." << LOG_ENDL;
        return nullptr;
    }

    // create resp_reader
    DataReader* resp_reader =
        dal_part_impl->CSSubscriber()->create_datareader(resp_topic_desc,
                                                         f_rqos,
                                                         client_impl->resp_reader_listener_,
                                                         StatusMask::subscription_matched());
    if (!resp_reader) {
        LOG_ERROR_WITH_LOCATION() << "create response reader failed." << LOG_ENDL;
        return nullptr;
    }

    client_impl->SetReqWriter(req_writer);
    client_impl->SetRespReader(resp_reader);

    // create dal_client
    dal_client_t* d_client = dal_client_allocate();
    if (!d_client) {
        LOG_ERROR_WITH_LOCATION() << "dal_client_allocate failed." << LOG_ENDL;
        return nullptr;
    }
    d_client->handle_ = client_impl;

    return d_client;
}

dal_client_t* dal_domain_participant_create_client(dal_domain_participant_t* participant,
                                                   const char* service_name,
                                                   const char* request_writer_type_name,
                                                   const char* response_reader_type_name,
                                                   dal_datawriter_qos_t* request_writer_qos,
                                                   dal_datareader_qos_t* response_reader_qos)
{
    DataWriterQos f_wqos;
    QosConverter::GetInstance()->ConvertDataWriterQosFromDalToFastdds(request_writer_qos, &f_wqos);
    f_wqos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_wqos.data_sharing().off();
#endif

    DataReaderQos f_rqos;
    QosConverter::GetInstance()->ConvertDataReaderQosFromDalToFastdds(response_reader_qos, &f_rqos);
    f_rqos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;
#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_rqos.data_sharing().off();
#endif

    return CreateDalClientByQos(participant,
                                service_name,
                                request_writer_type_name,
                                response_reader_type_name,
                                f_wqos,
                                f_rqos);
}

dal_client_t*
dal_domain_participant_create_client_with_profile(dal_domain_participant_t* participant,
                                                  const char* service_name,
                                                  const char* request_writer_type_name,
                                                  const char* response_reader_type_name,
                                                  dal_datawriter_qos_t* request_writer_qos,
                                                  dal_datareader_qos_t* response_reader_qos,
                                                  const char* request_writer_profile_name,
                                                  const char* response_reader_profile_name)
{
    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return nullptr;
    }

    const auto& dal_part_impl = reinterpret_cast<DalDomainParticipantImpl*>(participant->handle_);
    if (!dal_part_impl->CreateCSPubSub()) {
        LOG_ERROR_WITH_LOCATION() << "CreateCSPubSub failed." << LOG_ENDL;
        return nullptr;
    }

    DataWriterQos f_wqos;
    if (!request_writer_profile_name || strlen(request_writer_profile_name) == 0) {
        QosConverter::GetInstance()->ConvertDataWriterQosFromDalToFastdds(request_writer_qos,
                                                                          &f_wqos);
    } else {
        if (!ConfigManager::GetInstance()->getFastddsProfilePath().empty()) {
            try {
                dal_part_impl->CSPublisher()->get_datawriter_qos_from_profile(
                    request_writer_profile_name, f_wqos);
            }
            catch (...) {
                LOG_ERROR_WITH_LOCATION() << "get datawriter qos from profile failed." << LOG_ENDL;
                QosConverter::GetInstance()->ConvertDataWriterQosFromDalToFastdds(
                    request_writer_qos, &f_wqos);
            }
        }
    }
    f_wqos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_wqos.data_sharing().off();
#endif

    DataReaderQos f_rqos;
    if (!response_reader_profile_name || strlen(response_reader_profile_name) == 0) {
        QosConverter::GetInstance()->ConvertDataReaderQosFromDalToFastdds(response_reader_qos,
                                                                          &f_rqos);
    } else {
        if (!ConfigManager::GetInstance()->getFastddsProfilePath().empty()) {
            try {
                dal_part_impl->CSSubscriber()->get_datareader_qos_from_profile(
                    response_reader_profile_name, f_rqos);
            }
            catch (...) {
                LOG_ERROR_WITH_LOCATION() << "get datareader qos from profile failed." << LOG_ENDL;
                QosConverter::GetInstance()->ConvertDataReaderQosFromDalToFastdds(
                    response_reader_qos, &f_rqos);
            }
        }
    }
    f_rqos.endpoint().history_memory_policy =
        eprosima::fastrtps::rtps::MemoryManagementPolicy::PREALLOCATED_WITH_REALLOC_MEMORY_MODE;

#ifdef DAL_FASTDDS_DATA_SHARING_DISABLED
    f_rqos.data_sharing().off();
#endif

    return CreateDalClientByQos(participant,
                                service_name,
                                request_writer_type_name,
                                response_reader_type_name,
                                f_wqos,
                                f_rqos);
}

dal_ret_t dal_domain_participant_delete_client(dal_domain_participant_t* participant,
                                               dal_client_t* client)
{
    DomainParticipant* f_participant = get_fastdds_domain_participant(participant);
    if (!f_participant) {
        LOG_ERROR_WITH_LOCATION() << "get fastdds domain participant failed." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (!client || !client->handle_) {
        LOG_ERROR_WITH_LOCATION() << "client is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    auto client_impl = static_cast<DalClientImpl*>(client->handle_);
    if (!client_impl->Clear()) {
        LOG_ERROR_WITH_LOCATION() << "clear client failed." << LOG_ENDL;
        return DAL_RETCODE_ERROR;
    }
    delete client_impl;
    dal_client_free(client);

    return DAL_RETCODE_OK;
}

dal_ret_t dal_client_send_request(dal_client_t* client, void* request)
{
    if (!client || !client->handle_) {
        LOG_ERROR_WITH_LOCATION() << "client is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    auto client_impl = static_cast<DalClientImpl*>(client->handle_);

    WriteParams wparams;
    wparams.related_sample_identity().writer_guid() = client_impl->GetRespReader()->guid();

    dal_fastrtps_shared_cpp::SerializedData sdata;
    sdata.is_cdr_buffer = false;
    sdata.data = const_cast<void*>(request);
    sdata.impl = client_impl->req_ts_;

    if (client_impl->GetReqWriter()->write(&sdata, wparams)) {
        // not handle sequence_id
        return DAL_RETCODE_OK;
    } else {
        return DAL_RETCODE_ERROR;
    }
}

dal_ret_t dal_client_take_response(dal_client_t* client,
                                   dal_service_info_t* request_header,
                                   void* response,
                                   bool* taken)
{
    if (!client || !client->handle_) {
        LOG_ERROR_WITH_LOCATION() << "client is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (!request_header || !response) {
        LOG_ERROR_WITH_LOCATION() << "request_header or response is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (!taken) {
        LOG_ERROR_WITH_LOCATION() << "taken is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    *taken = false;

    auto client_impl = static_cast<DalClientImpl*>(client->handle_);

    ClientResponse resp;

    resp.buffer_.reset(new eprosima::fastcdr::FastBuffer());
    dal_fastrtps_shared_cpp::SerializedData sdata;
    sdata.is_cdr_buffer = true;
    sdata.data = resp.buffer_.get();
    sdata.impl = nullptr;   // not used when is_cdr_buffer is true

    eprosima::fastdds::dds::StackAllocatedSequence<void*, 1> data_values;
    const_cast<void**>(data_values.buffer())[0] = &sdata;
    eprosima::fastdds::dds::SampleInfoSeq info_seq{1};

    if (ReturnCode_t::RETCODE_OK == client_impl->GetRespReader()->take(data_values, info_seq, 1)) {
        if (info_seq[0].valid_data) {
            resp.sample_identity_ = info_seq[0].related_sample_identity;

            if (resp.sample_identity_.writer_guid() == client_impl->GetRespReader()->guid() ||
                resp.sample_identity_.writer_guid() == client_impl->GetReqWriter()->guid()) {
                auto raw_type_support = dynamic_cast<dal_fastrtps_shared_cpp::TypeSupport*>(
                    client_impl->resp_fastdds_type_.get());
                eprosima::fastcdr::Cdr deser(*resp.buffer_,
                                             eprosima::fastcdr::Cdr::DEFAULT_ENDIAN,
                                             eprosima::fastcdr::Cdr::DDS_CDR);
                if (raw_type_support->deserializeROSmessage(
                        deser, response, client_impl->resp_ts_)) {
                    request_header->source_timestamp = info_seq[0].source_timestamp.to_ns();
                    request_header->received_timestamp = info_seq[0].source_timestamp.to_ns();
                    request_header->request_id.sequence_number =
                        ((int64_t)resp.sample_identity_.sequence_number().high) << 32 |
                        resp.sample_identity_.sequence_number().low;

                    *taken = true;
                }
            }
        }
    }

    return DAL_RETCODE_OK;
}

void dal_client_set_on_response_callback(dal_client_t* client,
                                         dal_client_on_response_t callback,
                                         const void* user_data)
{
    if (!client || !client->handle_) {
        LOG_ERROR_WITH_LOCATION() << "client is null." << LOG_ENDL;
        return;
    }

    auto client_impl = static_cast<DalClientImpl*>(client->handle_);
    client_impl->resp_reader_listener_->set_on_new_response_callback(callback, user_data);
}

dal_ret_t dal_client_check_server_is_available(dal_client_t* client, bool* available)
{
    if (!client || !client->handle_) {
        LOG_ERROR_WITH_LOCATION() << "client is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    if (!available) {
        LOG_ERROR_WITH_LOCATION() << "available is null." << LOG_ENDL;
        return DAL_RETCODE_BAD_PARAMETER;
    }

    *available = false;

    auto client_impl = static_cast<DalClientImpl*>(client->handle_);
    
    size_t matched_request_writer_cnt = client_impl->req_writer_matched_count_.load();
    if (0 == matched_request_writer_cnt) {
        return DAL_RETCODE_OK;
    }

    size_t matched_responsed_reader_cnt = client_impl->resp_reader_matched_count_.load();
    if (0 == matched_responsed_reader_cnt) {
        return DAL_RETCODE_OK;
    }

    if (matched_request_writer_cnt != matched_responsed_reader_cnt) {
        return DAL_RETCODE_OK;
    }

    *available = true;
    return DAL_RETCODE_OK;
}